package net.bwie.douyin.dwd.log1.job;


import com.alibaba.fastjson.JSON;


import net.bwie.douyin.dwd.log1.Function.*;
import net.bwie.douyin.dwd.log1.bean.*;
import net.bwie.realtime.douyin.common.utils.JdbcUtil;
import net.bwie.realtime.douyin.common.utils.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;


import java.time.Duration;

/**
 * @ClassName：KafkaDataEtlJob
 * @Author: Vvn
 * @Date: 2025/5/27 9:28
 * @Description: 必须描述类做什么事情, 实现什么功能
 */

public class KafkaDataEtlJob1 {


    public static void main(String[] args) throws Exception {

        // 执行环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // DataStreamApi 读取  kafka数据
        DataStream<String> kafkaDataStream = KafkaUtil.consumerKafka(env, "live_events");

        // DataStream etl 过滤脏数据
        DataStream<String> etlDataStream = etlNoJsonData(kafkaDataStream);
        // 转换为实体类
        SingleOutputStreamOperator<TikTokBean> tiktokBeanStream = etlDataStream.process(new ProcessFunction<String, TikTokBean>() {

            @Override
            public void processElement(String value,
                                       ProcessFunction<String, TikTokBean>.Context ctx,
                                       Collector<TikTokBean> out) throws Exception {

                TikTokBean tikTokBean = JSON.parseObject(value, TikTokBean.class);
                out.collect(tikTokBean);
            }
        });

        // 设置水位线
        SingleOutputStreamOperator<TikTokBean> watermarks = tiktokBeanStream
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<TikTokBean>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                        .withTimestampAssigner(new SerializableTimestampAssigner<TikTokBean>() {
                            @Override
                            public long extractTimestamp(TikTokBean element, long recordTimestamp) {
                                return element.getTs();
                            }
                        }));

        // 根据直播间id分组
        KeyedStream<TikTokBean, String> KeyByStream = watermarks.keyBy(new KeySelector<TikTokBean, String>() {
            @Override
            public String getKey(TikTokBean value) throws Exception {
                return value.getRoomId();
            }
        });

        // 开窗
        WindowedStream<TikTokBean, String, TimeWindow> stream =
                KeyByStream.window(SlidingEventTimeWindows.of(Time.minutes(10), Time.minutes(1)));

        // 开窗聚合 实时求取在线人数--------------------------------------------------------------------------------------------------------
        DataStream<UserCountBean> UserCountStream = stream.apply(new realTimeUserCountFunction());
        DataStream<String> UserCountStringStream = UserCountStream.map(new MapFunction<UserCountBean, String>() {
            @Override
            public String map(UserCountBean value) throws Exception {
                return value.getWindowStartTime() +
                        "," + value.getWindowEndTime() +
                        "," + value.getRoomId() + ","
                        + value.getUserCount() + "," + value.getTs();
            }
        });
       // UserCountStringStream.print();
        JdbcUtil.sinkToClickhouseUpsert(UserCountStringStream, "insert into real_time_tiktok_report.tiktok_user_count values(?,?,?,?,?)");
        // 开窗聚合 实时求取礼物打赏----------------------------------------------------------------------------------------------------------
        DataStream<RoomGiftMoney> RoomGiftMoneyStream = stream.apply(new realTimeGiftFunction());
        DataStream<String> RoomGiftMoneyStringStream = RoomGiftMoneyStream.map(new MapFunction<RoomGiftMoney, String>() {
            @Override
            public String map(RoomGiftMoney value) throws Exception {
                return value.getWindowStartTime() + "," + value.getWindowEndTime() + "," + value.getRoomId() + "," + value.getGiftMoney() + "," + value.getTs();
            }
        });
       // RoomGiftMoneyStringStream.print();

        JdbcUtil.sinkToClickhouseUpsert(RoomGiftMoneyStringStream, "insert into real_time_tiktok_report.tiktok_gift_money values(?,?,?,?,?)");

        // 互动消息热度分析----------------------------------------------------------------------------------------------------------
        // 以直播间为主题 求取不同类型的 互动消息 如 share like message
        SingleOutputStreamOperator<RoomInteraction> interactionDataStream = stream.aggregate(new RealTimeInteractionOptimized.InteractionAggregator(), new RealTimeInteractionOptimized.InteractionWindowProcessor());
        DataStream<String> InteractionStringStream = interactionDataStream.map(new MapFunction<RoomInteraction, String>() {
            @Override
            public String map(RoomInteraction value) throws Exception {
                return value.getWindowStartTime() + "," + value.getWindowEndTime() +
                        "," + value.getRoomId() + "," + value.getInteractionCount() + ","
                        + value.getShareCount() + "," + value.getLikeCount() + "," +
                        value.getMessageCount() + "," + value.getTs();
            }
        });
        JdbcUtil.sinkToClickhouseUpsert(InteractionStringStream, "insert into real_time_tiktok_report.tiktok_interaction values(?,?,?,?,?,?,?,?)");
        // 刷礼物异常检测
        // 检测同一用户在 10秒内 累计送礼价值超过 10万元，通过 滚动窗口 聚合实现
//        WindowedStream<TikTokBean, String, TimeWindow> window =
//                watermarks.filter(event -> "gift".equals(event.getEventType()))
//                        .keyBy(TikTokBean::getUserId)
//                        .window(TumblingEventTimeWindows.of(Time.seconds(10)));
//
//        DataStream<AlertGiftSender> AlertGiftStream = window.apply(new realTimeGiftAlertFunction());
//        SingleOutputStreamOperator<AlertGiftSender> aggregate = window.aggregate(new GiftAmountAggregator(), new GiftAlertProcessFunction());
//        DataStream<String> AlertGiftStringStream = aggregate.map(new MapFunction<AlertGiftSender, String>() {
//                @Override
//            public String map(AlertGiftSender value) throws Exception {
//                return value.getUserId() + "," + value.getWindowStartTime() + "," + value.getWindowEndTime() + "," + value.getSendMoney() + "," + value.getTs();
//            }
//        });
       // AlertGiftStringStream.print();

        //JdbcUtil.sinkToClickhouseUpsert(AlertGiftStringStream, "insert into real_time_tiktok_report.tiktok_alert_gift values(?,?,?,?,?)");


        //机器人发言检测
//        WindowedStream<TikTokBean, String, TimeWindow> messageWindowStream = watermarks
//                // 1. 过滤发言事件
//                .filter(event -> "message".equals(event.getEventType()))
//                // 2. 按用户ID分组
//                .keyBy(TikTokBean::getUserId)
//                // 3. 定义滚动计数窗口
//                .window(TumblingEventTimeWindows.of(Time.seconds(10)));
//        DataStream<AlertMessageUser> messageStream = messageWindowStream.aggregate(new MessageCountAggregator(), new realTimeMessageCountFunction());
//
//        DataStream<String> streamOperator = messageStream.map(new MapFunction<AlertMessageUser, String>() {
//            @Override
//            public String map(AlertMessageUser value) throws Exception {
//                return value.getUserId() + "," + value.getWindowStartTime() + "," + value.getWindowEndTime() + "," + value.getMessageCount() + "," + value.getTs();
//            }
//        });


        //JdbcUtil.sinkToClickhouseUpsert(streamOperator, "insert into real_time_tiktok_report.tiktok_alert_message values(?,?,?,?,?)");

        env.execute("KafkaDataEtlJob");
    }

    private static DataStream<String> etlNoJsonData(DataStream<String> kafkaDataStream) {

        OutputTag<String> err = new OutputTag<String>("err") {
        };

        // DataStream etl 过滤脏数据
        DataStream<String> etlDataStream = kafkaDataStream.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String value,
                                       ProcessFunction<String, String>.Context ctx,
                                       Collector<String> out) throws Exception {
                try {

                    JSON.parseObject(value);
                    out.collect(value);
                } catch (Exception e) {
                    ctx.output(err, value);
                }
            }
        });
        return etlDataStream;
    }


}
